package com.lou1052.redis.jedis.source.share;

import com.lou1052.core.base.annotation.thread.ThreadSafe;
import com.lou1052.redis.config.ConfigType;
import com.lou1052.redis.config.NodeConfig;
import com.lou1052.redis.jedis.config.JedisConfig;
import com.lou1052.redis.jedis.config.PoolConfig;
import com.lou1052.redis.jedis.ext.client.shard.ShardedJedisExt;
import com.lou1052.redis.jedis.ext.client.shard.ShardedJedisExtPool;
import com.lou1052.redis.jedis.source.AbsJedisSource;
import com.lou1052.redis.jedis.util.SerSelector;
import org.apache.commons.pool2.PooledObject;
import redis.clients.jedis.JedisShardInfo;

import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.function.Supplier;

/**
 * shard redis ext
 * Created with IntelliJ IDEA.
 * User: liuqing
 * Date: 2015/10/14
 * Time: 22:25
 * To change this template use File | Settings | File Templates.
 */
@ThreadSafe
public class ShardedJedisExtSource extends AbsJedisSource<ShardedJedisExt> {

    private ShardedJedisExtPool shardedJedisPool;

    /**
     * 连接池使用的Deque实现
     */
    private final Supplier<Deque<PooledObject<ShardedJedisExt>>> dequeSupplier;

    public ShardedJedisExtSource(JedisConfig jedisConfig, SerSelector serSelectors, boolean concurrent) {
        super(jedisConfig,serSelectors);
        if(concurrent){
            dequeSupplier = ConcurrentLinkedDeque::new;
        }else{
            dequeSupplier=null;//use default
        }
    }

    @Override
    public ShardedJedisExt getSource(){
        ShardedJedisExt shardedJedis = shardedJedisPool.getResource();
        shardedJedis.setSerializerFunction(serializerFunction);
        return shardedJedis;
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        if(jedisConfig.getConfigType()!= ConfigType.SHARE){
            throw  new IllegalArgumentException("Jedis share init but config is not ConfigType.SHARE");
        }
        PoolConfig poolConfig = jedisConfig.getPoolConfig();
        NodeConfig nodeConfig = jedisConfig.getNodeConfig();
        List<JedisShardInfo> shards = new ArrayList<>();
        //todo add param timeout 2000
        nodeConfig.getNodeInfoSet().forEach((nodeInfo -> {shards.add(
                new JedisShardInfo(nodeInfo.getIp(),nodeInfo.getName(),nodeInfo.getPort(),2000,nodeInfo.getWeight()));}));
        shardedJedisPool = new ShardedJedisExtPool(poolConfig,shards,dequeSupplier);
        logger.debug("init share Jedis clients");
    }


    @Override
    public void destroy() throws Exception {
        shardedJedisPool.destroy();
    }
}
